package com.zhang.hadoop.flink.test11;

import com.zhang.hadoop.flink.test10.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author: zhang yufei
 * @createTime:2022/12/10 13:59
 * @description:
 */
public class OrderTimeoutDetectExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.获取数据流
        SingleOutputStreamOperator<OrderEvent> orderEventStream = env.fromElements(
                new OrderEvent("ouyanghui", "yindao", "mo", 2000L),
                new OrderEvent("chenyuping", "gangmen", "mo", 3000L),
                new OrderEvent("ouyanghui", "yindao", "she", 4000L),
                new OrderEvent("ouyanghui", "yindao", "cha", 5000L),
                new OrderEvent("chenyuping", "niaodao", "mo", 8000L),
                new OrderEvent("chenyuping", "niaodao", "cha", 9000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                    @Override
                    public long extractTimestamp(OrderEvent orderEvent, long l) {
                        return orderEvent.timestamp;
                    }
                }));

        //2.定义模式
        Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("mo")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return orderEvent.eventType.equals("mo");
                    }
                })
                .followedBy("cha")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return orderEvent.eventType.equals("cha");
                    }
                })
                .within(Time.minutes(15));

        //3.将模式应用到订单数据流上
        PatternStream<OrderEvent> patternStream = CEP.pattern(orderEventStream.keyBy(event -> event.orderId), pattern);

        //4.定义一个侧输出流标签
        OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {};

        //5.将完全匹配和超时部分匹配的复试事件
        SingleOutputStreamOperator<String> result = patternStream.process(new OrderPayMatch());

        //打印输出
        result.print("payed:");
        result.getSideOutput(timeoutTag).print("timeout:");

        env.execute();
    }

    public static class OrderPayMatch extends PatternProcessFunction<OrderEvent, String> implements TimedOutPartialMatchHandler<OrderEvent> {

        @Override
        public void processMatch(Map<String, List<OrderEvent>> map, Context context, Collector<String> collector) throws Exception {
            //获取当前的支付事件
            OrderEvent payEvent = map.get("cha").get(0);
            collector.collect("用户" + payEvent.userId + "订单" + payEvent.orderId + "已支付！");
        }

        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> map, Context context) throws Exception {
            OrderEvent creatEvent = map.get("mo").get(0);
            OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {};
            context.output(timeoutTag, "用户" + creatEvent.userId + "订单" + creatEvent.orderId + "超时未支付！");
        }
    }
}
